using KafkaFlow;
using KafkaFlow.Consumers;
using KafkaFlow.Serializer;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

namespace Application.KafkaService;

public class KafkaConsumerService : IHostedService
{
    // private readonly IConsumer<Ignore, string> _consumer;
    //
    // public KafkaConsumerService()
    // {
    //     var config = new ConsumerConfig
    //     {
    //         BootstrapServers = "localhost:9092",
    //         GroupId = "test-consumer-group",
    //         AutoOffsetReset = AutoOffsetReset.Earliest,
    //         EnableAutoCommit = false,
    //         StatisticsIntervalMs = 5000,
    //     };
    //
    //     _consumer = new ConsumerBuilder<Ignore, string>(config).Build();
    // }


    public Task StartAsync(CancellationToken cancellationToken)
    {
        // 启动消费者线程
        Task.Run(() => ConsumeMessages("test-topic",cancellationToken), cancellationToken);

        Console.WriteLine("开始消费");
        
        return Task.CompletedTask;
    }

    private async Task ConsumeMessages(string topic,CancellationToken cancellationToken)
    {
        // _consumer.Subscribe(topic);
        //
        // while (!cancellationToken.IsCancellationRequested)
        // {
        //     try
        //     {   
        //         var consumeResult = _consumer.Consume(cancellationToken);
        //         Console.WriteLine($"Consumed message: {consumeResult.Message.Value} at {consumeResult.TopicPartitionOffset}");
        //     }
        //     catch (ConsumeException e)
        //     {
        //         Console.WriteLine($"Error: {e.Error.Reason}");
        //     }
        // }
        
        var services = new ServiceCollection();
        
        services.AddKafka(kafka => kafka
            .AddCluster(cluster => cluster
                .WithBrokers(new[] { "localhost:9092" })
                .AddConsumer(consumer => consumer
                    .Topic(topic)
                    .WithGroupId("sample-group")
                    .WithBufferSize(100)
                    .WithWorkersCount(10)
                    .AddMiddlewares(middlewares => middlewares.AddDeserializer<JsonCoreDeserializer>().AddBatching(20, TimeSpan.FromSeconds(5)).Add<HandlingMiddleware>()
                    )
                )
            )
        );
        
        var provider = services.BuildServiceProvider();

        var bus = provider.CreateKafkaBus();

        await bus.StartAsync(cancellationToken);
       
        
    }

    public Task StopAsync(CancellationToken cancellationToken)
    {
        //_consumer.Close();
        
        Console.WriteLine("消费已停止");
        
        return Task.CompletedTask;
    }
}


internal class HandlingMiddleware : IMessageMiddleware
{
    public Task Invoke(IMessageContext context, MiddlewareDelegate next)
    {
        var batch = context.GetMessagesBatch();

        Console.WriteLine($"批次 {batch.Count}");
        
        // foreach (var message in batch)
        // {
        //     Console.WriteLine($"Consumed message: {message.Message.Value}");
        // }

        return Task.CompletedTask;
    }
}